package com.test.myrxjava.core;

public class ObservableCreate<T> extends Observable<T> {//创建被观察者类


    final ObservableOnSubscribe<T> source;//被观察者被订阅

    public ObservableCreate(ObservableOnSubscribe<T> source) {
        this.source = source;
    }

    @Override
    protected void subscribeActual(Observer observer) {//订阅时
        observer.onSubscribe();//回调订阅回调告知方法
        CreateEmitter<T> emitter = new CreateEmitter<>(observer);//创建发射器
        source.subscribe(emitter);//被观察者被订阅回调
    }

    static class CreateEmitter<T> implements Emitter<T>{//具体发射器的接口实现

        Observer<T> observer;
        boolean done;

        public CreateEmitter(Observer<T> observer){
            this.observer = observer;
        }

        @Override
        public void onNext(T t) {
            if(done)
                return;
            observer.onNext(t);
        }

        @Override
        public void onComplete() {
            if(done)
                return;
            observer.onComplete();
        }

        @Override
        public void onError(Throwable throwable) {
            if(done)
                return;
            observer.onError(throwable);
        }
    }
}
